package offset_gaobo.util;

import offset_gaobo.Const;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import zookeeper.curator.demo._02_create_node;

import java.util.List;

// 连接 zookeeper 用于存放 oofset
public class ZookeeperUtils {


    public static CuratorFramework client = null;
    public static final String zkServerPath = "zocoo:2181";

    //  无参构造器
    // 实例化 zookeeper客户端
    static {
        // 重连3次 每次间隔5秒
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        // 建造者模式 返回一个连接指定zookeeper的对象
        client = CuratorFrameworkFactory
                .builder()
                .connectString(zkServerPath)
                .sessionTimeoutMs(10000)
                .retryPolicy(retryPolicy)
                .namespace(Const.ZOOKEEPER_NAME_SPACE)
                .build();
        client.start();
    }


    // 关闭zk连接
    public void closeZKClient() {
        if (client != null) {
            client.close();
        }
    }

    /**
     * 获取Stat
     * @param nodePath
     * @return
     * @throws Exception
     */
    public static Stat getStat(String nodePath) throws Exception {

        ZookeeperUtils zookeeperUtils = new ZookeeperUtils();
        Stat stat = new Stat();
        zookeeperUtils.client.getData().storingStatIn(stat).forPath(nodePath);

        return stat;
    }

    /**
     * 判断该路径/节点是否存在
     * @param nodePath 节点路径
     * @return
     * @throws Exception
     */
    public static boolean checkExists(String nodePath) throws Exception {
        Stat statExist = client.checkExists().forPath(nodePath);
        if (statExist == null){
            return false;
        }else {
            return true;
        }
    }


    /**
     * 创建zookeeper节点,并存储数据
     * @param nodePath
     * @param data
     * @throws Exception
     */
    public static void createNode( String nodePath,String data) throws Exception {
        byte[] dataByte = data.getBytes();
        client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath(nodePath, dataByte);
    }


    /**
     * 获取子节点路径
     * @param nodePath
     * @return
     */
    public static List<String> getChildren( String nodePath)  {
        List<String> childNodes = null;
        try {
            childNodes = client.getChildren()
                    .forPath(nodePath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return childNodes;
    }

    /**
     *
     * @param offsetPath
     * @param offsetJson
     * @param version
     * @throws Exception
     */
    public static void updataOffset(String offsetPath,String offsetJson,int version) throws Exception {
        byte[] newdata=offsetJson.getBytes();
        client.setData().withVersion(version).forPath(offsetPath,newdata);

    }

    /**
     *
     * @param nodePath
     * @return
     * @throws Exception
     */
    public static String getDate(String nodePath){
        byte[] data = new byte[0];
        try {
            data = client.getData().forPath(nodePath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new String(data);
    }



}
